﻿﻿// Copyright (c) Microsoft Corporation. All rights reserved. See License.txt in the project root for license information.

using System;
using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;

namespace Nardax.Azure.ServiceBus
{
    public class TopicAdapter : ITopicAdapter
    {
        private readonly ITopicClientFactory _topicClientFactory;
        private bool _isInitialized;
        private readonly object _initializationLock;
        private readonly IMessageSender[] _messageSenders;
        private readonly object _startingIndexLock;
        private int _startingIndex;

        public TopicAdapter(TopicAdapterDescription topicAdapterDescription)
            : this(topicAdapterDescription, new TopicClientFactory())
        {
        }

        public TopicAdapter(TopicAdapterDescription topicAdapterDescription, ITopicClientFactory topicClientFactory)
        {
            if (topicAdapterDescription == null)
            {
                throw new ArgumentNullException("topicAdapterDescription");
            }

            if (topicClientFactory == null)
            {
                throw new ArgumentNullException("topicClientFactory");
            }

            TopicAdapterDescription = topicAdapterDescription;
            _topicClientFactory = topicClientFactory;
            _initializationLock = new object();
            _messageSenders = new IMessageSender[TopicAdapterDescription.ConnectionStrings.Length];
            _startingIndexLock = new object();
        }

        public void Initialize()
        {
            try
            {
                InitializeInternal();
            }
            catch (Exception ex)
            {
                var initializationException = new ServiceBusAdapterInitializationException("TopicAdapter failed to initialize.", ex);
                throw initializationException;
            }
        }

        private void InitializeInternal()
        {
            if (_isInitialized)
            {
                return;
            }

            lock (_initializationLock)
            {
                if (_isInitialized)
                {
                    return;
                }

                for (var i = 0; i < TopicAdapterDescription.ConnectionStrings.Length; i++)
                {
                    TryCreateTopic(i);
                    _messageSenders[i] = CreateTopicClient(i);
                }

                _isInitialized = true;
            }
        }

        private void TryCreateTopic(int index)
        {
            var namespaceManager = NamespaceManager.CreateFromConnectionString(TopicAdapterDescription.ConnectionStrings[index]);
            if (!TopicAdapterDescription.RetryPolicy.ExecuteAction(() => namespaceManager.TopicExists(TopicAdapterDescription.TopicDescription.Path)))
            {
                try
                {
                    TopicAdapterDescription.RetryPolicy.ExecuteAction(() => namespaceManager.CreateTopic(TopicAdapterDescription.TopicDescription));
                }
                catch (MessagingEntityAlreadyExistsException)
                {
                    // I am handling this exception because it is telling me that the topic I am trying to create already
                    // exists. This is possible because other adapters may have created it just after I checked to see
                    // if it was already there.
                }
            }
        }

        private IMessageSender CreateTopicClient(int index)
        {
            var topicClient = _topicClientFactory.CreateTopicClient(TopicAdapterDescription.ConnectionStrings[index], TopicAdapterDescription.TopicDescription.Path);
            return topicClient;
        }

        public bool TrySend<T>(T message) where T : IBrokeredMessageContainer
        {
            return TrySend(message.BrokeredMessage);
        }

        public bool TrySend(BrokeredMessage message)
        {
            if (message == null)
            {
                throw new ArgumentNullException("message");
            }

            Initialize();

            // increment the starting index to achieve a round-robin effect when talking to multiple topic clients
            IncrementStartingIndex();

            // get a local copy of the starting index so that I can iterate it independently of other calls to TrySend
            // and ensure that the round robin effect is touching all of the available clients per call if needed
            var index = _startingIndex;

            for (int i = 0; i < _messageSenders.Length; i++)
            {
                int retryCount = -1;

                try
                {
                    var topicClient = _messageSenders[index];
                    TopicAdapterDescription.RetryPolicy.ExecuteAction(() => {
                                                                                retryCount++;
                                                                                topicClient.Send(message);
                                                                            });
                    return true;
                }
                catch (Exception ex)
                {
                    // if this exception has been raised while a retry is in effect that means than the exception was encountered because of the retry
                    if (ex is InvalidOperationException && retryCount > 0)
                    {
                        return true;
                    }

                    OnSendFailed(_messageSenders[index], ex);

                    if (ex is OperationCanceledException)
                    {
                        // Client entity has been closed or aborted so I am going to replace it
                        _messageSenders[index] = CreateTopicClient(index);
                    }

                    IncrementIndex(ref index);
                }
            }

            return false;
        }

        private void IncrementStartingIndex()
        {
            lock (_startingIndexLock)
            {
                IncrementIndex(ref _startingIndex);
            }
        }

        private void IncrementIndex(ref int index)
        {
            index = (++index % TopicAdapterDescription.ConnectionStrings.Length);
        }

        private void OnSendFailed(IMessageSender messageSender, Exception exception)
        {
            if (SendFailed != null)
            {
                var sendFailedEventArgs = new SendFailedEventArgs(exception);
                sendFailedEventArgs.Namespace = messageSender.Namespace;
                sendFailedEventArgs.TopicPath = messageSender.Path;

                SendFailed(this, sendFailedEventArgs);
            }
        }

        public event EventHandler<SendFailedEventArgs> SendFailed;

        public TopicAdapterDescription TopicAdapterDescription { get; private set; }
    }
}
